package com.mjf.day6

import java.util
import java.util.concurrent.TimeUnit

import com.mjf.day3.SensorReading
import org.apache.flink.api.common.functions.{ReduceFunction, RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state._
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 1.checkpoint:
 *   >保存的是所有任务 状态 的快照
 *   >状态要求是 所有任务都处理完 同一个数据之后的状态
 *
 * 2.flink checkpoint 机制原理:
 *   基于 Chandy-Lamport 算法
 *
 * 3.存储位置:
 *   状态后端的位置
 *
 * 4.flink checkpoint 流程:
 *   >每个需要checkpoint的应用在启动时，flink的JobManager为其创建一个CheckPointCoordinator(检查点协调器)
 *   >CheckPointCoordinator负责本应用快照的制作
 *   1）CheckPointCoordinator周期性的向该流的所有source算子发送barrier(屏障)
 *   2）当某个source算子收到barrier时，便暂停数据处理过程。
 *      然后将自己的当前状态制作成快照，并保存到指定的持久化存储中。
 *      最后向CheckPointCoordinator报告自己的快照制作情况，同时向自身下游所有算子广播该barrier，恢复数据处理。
 *   3）下游算子收到barrier之后，会暂停自己的数据处理过程。
 *      然后将自己的当前状态制作成快照，并保存到指定的持久化存储中。
 *      最后向CheckPointCoordinator报告自己的快照制作情况，同时向自身下游所有算子广播该barrier，恢复数据处理。
 *   4）每个算子都按照步骤三不断制作快照并向下游广播，直到最后barrier传递到sink算子，快照制作完成。
 *   5）当CheckPointCoordinator收到所有算子的快照之后，认为该周期的快照制作成功；
 *      否则，如果在规定的时间内没有收到所有算子的报告，则认为本周期的快照制作失败。
 *   >如果一个算子有两个输入源，则暂时阻塞先收到barrier的数据源
 *   >等第两个编号相同的barrier到来时，再制作自身快照并向下游广播该barrier。
 *
 * 5.flink 重启策略:
 *   >固定延迟重启策略(Fixed Delay Restart Strategy)
 *   >失败率重启策略(Failure Rate Restart Strategy)
 *   >无重启策略
 *
 */
object CheckPointExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 配置状态后端
//    env.setStateBackend(new MemoryStateBackend()) // 内存级的状态后端,存储在TaskManager的JVM堆上
//    env.setStateBackend(new FsStateBackend("")) // 存到远程的FileSystem上，本地状态跟MemoryStateBackend一样
//    env.setStateBackend(new RocksDBStateBackend("", true))  // 将所有状态序列化后，存入本地的RocksDB中存储


    // checkpoint相关配置
    // 启用检查点，指定触发检查点的间隔时间(毫秒)
    env.enableCheckpointing(1000L)
    // 其他配置
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    env.getCheckpointConfig.setCheckpointTimeout(30000L)
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500L) // 本次检查点结束到下次检查点开始最小间隔时间
    env.getCheckpointConfig.setPreferCheckpointForRecovery(true)  // 设置作业恢复是否应在有更近保存点时回退到检查点
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3)


    // 重启策略的配置
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L))
    env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.of(5, TimeUnit.SECONDS), Time.of(10, TimeUnit.SECONDS)))


    val inputStream: DataStream[SensorReading] =
      env.socketTextStream("hadoop103", 9999)
        .map {
          line =>
            val strings: Array[String] = line.split(",")
            SensorReading(strings(0), strings(1).toLong, strings(2).toDouble)
        }

    inputStream.map(new MyMap())  // operator state

    val warningStream: DataStream[(String, Double, Double)] = inputStream
      .keyBy("id")
      .flatMap(new TempChangeWarningWithFlatMap(10))  // keyed state

    warningStream.print()

    env.execute("StateExample")

  }
}


// 自定义 RichMapFunction
class TempChangeWarning(threshold: Double) extends RichMapFunction[SensorReading, (String, Double, Double)]{

  // 定义状态变量，上一次的温度值
  var lastTempState: ValueState[Double] = _

  override def open(parameters: Configuration): Unit = {
    lastTempState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("last-temp", classOf[Double]))
  }

  override def map(value: SensorReading): (String, Double, Double) = {
    // 从状态中取出上次的温度值
    val lastTemp: Double = lastTempState.value()
    // 更新状态
    lastTempState.update(value.temperature)

    // 跟当前温度值计算差值，然后跟阈值比较，如果大于就报警
    val diff: Double = (value.temperature - lastTemp).abs
    if (diff > threshold) {
      (value.id, lastTemp, value.temperature)
    } else {
      null  // 会报空指针异常
    }
  }

}

// 自定义 RichFlatMapFunction,可以输出多个结果
class TempChangeWarningWithFlatMap(threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {

  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("last-temp", classOf[Double]))

  override def flatMap(value: SensorReading, out: Collector[(String, Double, Double)]): Unit = {
    val lastTemp: Double = lastTempState.value()
    lastTempState.update(value.temperature)

    val diff: Double = (lastTemp - value.temperature).abs
    if(diff > threshold) {
      out.collect((value.id, lastTemp, value.temperature))
    }
  }

}


// keyed state定义示例
class MyProcessor extends KeyedProcessFunction[String, SensorReading, Int] {

  // 由于需要用到上下文环境`getRuntimeContext`,在程序编译的时候是没有上下文环境的。
  // 所以使用懒加载的方式来创建
//  lazy val myState: ValueState[Int] = getRuntimeContext.getState(new ValueStateDescriptor[Int]("my-state", classOf[Int]))

  var myState: ValueState[Int] = _
  lazy val myListState: ListState[String] = getRuntimeContext.getListState(new ListStateDescriptor[String]("my-list-state", classOf[String]))
  lazy val myMapState: MapState[String, Double] = getRuntimeContext.getMapState(new MapStateDescriptor[String, Double]("my-map-state", classOf[String], classOf[Double]))
  lazy val myReducingState: ReducingState[SensorReading] = getRuntimeContext.getReducingState(
    new ReducingStateDescriptor[SensorReading](
      "my-reducing-state",
      new ReduceFunction[SensorReading]() {
        override def reduce(value1: SensorReading, value2: SensorReading): SensorReading = {
          SensorReading(value1.id, value1.timestamp.max(value2.timestamp), value1.temperature.max(value2.temperature))
        }
      },
      classOf[SensorReading]))

  // 由于需要用到上下文环境`getRuntimeContext`,在程序编译的时候是没有上下文环境的。
  // 所以需要在初始化方法中创建（确保上下文环境存在以后）。效果同上
  override def open(parameters: Configuration): Unit = {
    myState = getRuntimeContext.getState(new ValueStateDescriptor[Int]("my-state", classOf[Int]))
  }

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, Int]#Context, out: Collector[Int]): Unit = {
    myState.value()
    myState.update(1)

    myListState.get()
    myListState.add("1")
    myListState.addAll(new util.ArrayList[String]())
    myListState.update(new util.ArrayList[String]())

    myMapState.get("sensor_1")
    myMapState.put("sensor_1", 10.1)

    myReducingState.add(value)
    myReducingState.clear()
  }

}

// operator state示例
class MyMap extends RichMapFunction[SensorReading, Long] with ListCheckpointed[java.lang.Long]{

  // 直接使用本地变量没有办法故障恢复
  var count: Long = 0L

  override def map(value: SensorReading): Long = {
    count += 1
    count
  }

  override def snapshotState(checkpointId: Long, timestamp: Long): util.List[java.lang.Long] = {
    val stateList: util.ArrayList[java.lang.Long] = new util.ArrayList[java.lang.Long]()
    stateList.add(count)
    stateList
  }

  override def restoreState(state: util.List[java.lang.Long]): Unit = {
    val iter: util.Iterator[java.lang.Long] = state.iterator()
    if (iter.hasNext) {
      count += iter.next()
    }
  }

}